package com.warm.pump.executor.batch;

import com.warm.pump.frame.db.DbUtil;
import com.warm.pump.frame.gen.util.TypeCast;
import com.warm.pump.frame.util.SpringBeanUtil;
import com.warm.pump.module.work.bean.po.gen.WorkItem;
import com.warm.pump.module.work.bean.po.gen.WorkItemSub;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 中间转换器
 * Created by guozj on 2016/12/21.
 */
public class CursorProcessor implements ItemProcessor<Object, Object> {
    //查询
    private static final Logger log = LoggerFactory.getLogger(CursorProcessor.class);
    @Autowired
    private RedisTemplate redisTemplate;
    private boolean isFirst = true;
    private Map<String, Object> condValueMap = new HashMap<>();
    private WorkItem itemDtl;
    private List<WorkItemSub> subList;


    /**
     * @see ItemProcessor#process(Object)
     */
    public Object process(Object object) throws Exception {

        //删除操作
        if (isFirst) {
            if (StringUtils.isNotEmpty(itemDtl.getDeleteSql())) {
                deleteDeal();
            }
            if ("REDIS_M".equals(itemDtl.getDetailMode()) &&
                    (StringUtils.isEmpty(itemDtl.getCacheType()) ||
                            "all".equals(itemDtl.getCacheType()))) {
                removeRedis();
            }
        }

        isFirst = false;

        if (null != object) {
            Map obMap = (Map) object;
            String processClassType = itemDtl.getProcessClassType();

            Map rtMap = new HashMap<String, Object>();
            rtMap.putAll(obMap);
            //增量数据，对历史数据进行处理
            if (StringUtils.isNotEmpty(itemDtl.getSelectOne())) {
                Map oneMap = DbUtil.findOne(itemDtl.getDbKeyTo(), itemDtl.getSelectOne(), obMap);
                if (null != oneMap && oneMap.size() > 0) {
                    if (StringUtils.isNotEmpty(itemDtl.getUpdateSql())) {
                        DbUtil.update(itemDtl.getDbKeyTo(), itemDtl.getUpdateSql(), oneMap);
                    }
                    rtMap.putAll(oneMap);
                }
            }
            //处理子模块
            if (this.subList != null) {
                final Map<String, Object> fobMap = obMap;
                List<Future<Map<String, Object>>> futureList = new ArrayList<>();
                ExecutorService executorService = Executors.newFixedThreadPool(10);
                for (WorkItemSub sub : subList) {

                    //通过callable线程 把每个子模块
                    Future<Map<String, Object>> future = executorService.submit(() -> {
                        Map<String, Object> callMap = new HashMap<>();

                        if (StringUtils.isNotEmpty(sub.getSubType()) && "SQL".equals(sub.getSubType())) {
                            String subDbKey = sub.getDbKeyFrom();
                            String subSql = sub.getSubSql();

                            if (StringUtils.isEmpty(subDbKey)) {
                                subDbKey = itemDtl.getDbKeyFrom();
                            }
                            Map<String, Object> rtnMap = DbUtil.findOne(subDbKey, subSql, fobMap);
                            if (rtnMap != null) {
                                callMap.putAll(rtnMap);
                            }
                        } else if (StringUtils.isNotEmpty(sub.getSubType()) && "REDIS".equals(sub.getSubType())) {
                            String cacheName = sub.getCacheName();
                            if (StringUtils.isEmpty(sub.getCacheKeyField())) {
                                return null;
                            }
                            String[] cacheKeyFields = sub.getCacheKeyField().split(",");
                            String cacheGetFields = sub.getCacheGetFields();
                            if (StringUtils.isNotEmpty(cacheName) && StringUtils.isNotEmpty(cacheGetFields)) {
                                String[] getFields = cacheGetFields.split(",");

                                StringBuilder cacheKey = new StringBuilder();
                                int j = 1;
                                for (String cacheField : cacheKeyFields) {
                                    if (StringUtils.isNotEmpty(cacheField)) {
                                        cacheKey.append(TypeCast.getString(fobMap.get(cacheField)));
                                        if (j < cacheKeyFields.length) {
                                            cacheKey.append("-");
                                        }
                                    }
                                    j++;
                                }
                                if (StringUtils.isEmpty(cacheKey.toString())) {
                                    return null;
                                }
                                Map<String, Object> rdMap = (Map<String, Object>) redisTemplate.opsForHash().get(cacheName, cacheKey.toString());

                                if (null != rdMap && rdMap.size() > 0) {
                                    for (String field : getFields) {
                                        callMap.put(field, rdMap.get(field));
                                    }
                                }
                            }
                        }
                        return callMap;
                    });
                    futureList.add(future);
                }

                //关闭线程池
                executorService.shutdown();
                //通过Future可以并发发出N个请求，然后等待最慢的一个返回
                for (Future<Map<String, Object>> future : futureList) {
                    try {
                        Map<String, Object> returnMap = future.get();
                        if (null != returnMap) {
                            rtMap.putAll(returnMap);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            }


            //处理类
            if ((StringUtils.isEmpty(processClassType) || "javaClass".equals(processClassType))
                    && StringUtils.isNotEmpty(itemDtl.getProcessClass())) {
                BaseProcessService processService = (BaseProcessService) Class.forName(itemDtl.getProcessClass()).newInstance();
                processService.initParams(itemDtl);
                rtMap = processService.deal(rtMap);
                if (null == rtMap) {
                    return null;
                }
            }

            if ((StringUtils.isEmpty(processClassType) || "springBean".equals(processClassType))
                    && StringUtils.isNotEmpty(itemDtl.getProcessBean())) {
                BaseProcessService processService = SpringBeanUtil.getBean(itemDtl.getProcessBean());
                processService.initParams(itemDtl);
                rtMap = processService.deal(rtMap);
                if (null == obMap) {
                    return null;
                }
            }

            object = rtMap;

        }
        return object;
    }

    /**
     * 删除处理 加同步锁，用isFirst标识保证只有第一个线程 执行删除操作
     */
    private synchronized void deleteDeal() {
        if (this.isFirst) {

            String sqlDelete = itemDtl.getDeleteSql();

            log.debug("执行删除:" + sqlDelete);
            DbUtil.delete(itemDtl.getDbKeyTo(), sqlDelete, condValueMap);
            //是否是第一个线程 只能放到最后
            this.isFirst = false;
        }
    }


    private synchronized void removeRedis() {
        if (this.isFirst) {

            if (redisTemplate.hasKey(itemDtl.getCacheName())) {
                redisTemplate.delete(itemDtl.getCacheName());
            }
            //是否是第一个线程 只能放到最后
            this.isFirst = false;
        }
    }

    public Map<String, Object> getCondValueMap() {
        return condValueMap;
    }

    public void setCondValueMap(Map<String, Object> condValueMap) {
        this.condValueMap = condValueMap;
    }


    public void setItemDtl(WorkItem itemDtl) {
        this.itemDtl = itemDtl;
    }

    public void setSubList(List<WorkItemSub> subList) {
        this.subList = subList;
    }
}
